package com.datastax.driver.mapping;

import com.datastax.driver.core.*;
import com.google.common.base.Function;
import com.google.common.base.Functions;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.Uninterruptibles;

import java.util.*;
import java.util.concurrent.ExecutionException;

/**
 * Created by dave on 18-7-26 上午9:45.
 */
public class CassandraMapper<T> extends Mapper<T> {
    private static final Function<Object, Void> TO_NULL = Functions.constant(null);
    private final MappingManager manager;
    private final Class<T> klass;
    private final EntityMapper<T> mapper;

    CassandraMapper(MappingManager manager, Class<T> klass, EntityMapper<T> mapper) {
        super(manager, klass, mapper);
        this.manager = manager;
        this.klass = klass;
        this.mapper = mapper;
    }

    public void save(List<T> list) {
        checkNotInEventLoop();
        try {
            Uninterruptibles.getUninterruptibly(saveAsync(list));
        } catch (ExecutionException e) {
            throw DriverThrowables.propagateCause(e);
        }
    }

    public void save(List<T> list, Option... options) {
        checkNotInEventLoop();
        try {
            Uninterruptibles.getUninterruptibly(saveAsync(list, options));
        } catch (ExecutionException e) {
            throw DriverThrowables.propagateCause(e);
        }
    }

    public ListenableFuture<Void> saveAsync(List<T> list) {
        return submitVoidQueryAsync(saveQueryAsync(list, new EnumMap<>(Option.Type.class)));
    }

    public ListenableFuture<Void> saveAsync(List<T> list, Option... options) {
        return submitVoidQueryAsync(saveQueryAsync(list, toMap(options)));
    }

    private ListenableFuture<Void> submitVoidQueryAsync(ListenableFuture<Statement> bsFuture) {
        ListenableFuture<ResultSet> rsFuture = GuavaCompatibility.INSTANCE.transformAsync(bsFuture, (Statement bs) -> session().executeAsync(bs));
        return Futures.transform(rsFuture, TO_NULL);
    }

    private ListenableFuture<Statement> saveQueryAsync(List<T> list, final EnumMap<Option.Type, Option> options) {
        TreeMap<AliasedMappedProperty, Object> columns = new TreeMap<>();
        for (AliasedMappedProperty col : mapper.allColumns) {
            if (!col.mappedProperty.isComputed()) {
                columns.put(col, true);
            }
        }

        return Futures.transform(getPreparedQueryAsync(QueryType.SAVE, columns.keySet(), options), (PreparedStatement input) -> {
            BatchStatement batch = new BatchStatement(BatchStatement.Type.UNLOGGED); // 不必加事务
            for (T entity : list) {
                BoundStatement bs = input.bind();
                int i = 0;
                for (AliasedMappedProperty col : mapper.allColumns) {
                    Object value = col.mappedProperty.getValue(entity);
                    if (!col.mappedProperty.isComputed()) {
                        columns.put(col, value);
                    }
                }
                for (Map.Entry<AliasedMappedProperty, Object> entry : columns.entrySet()) {
                    AliasedMappedProperty mapper = entry.getKey();
                    Object value = entry.getValue();
                    setObject(bs, i++, value, mapper);
                }

                if (mapper.writeConsistency != null)
                    bs.setConsistencyLevel(mapper.writeConsistency);

                for (Option option : options.values()) {
                    option.validate(QueryType.SAVE, manager);
                    i = option.apply(bs, i);
                }
                batch.add(bs);
            }

            return batch;
        });
    }

    private static <T> void setObject(BoundStatement bs, int i, T value, AliasedMappedProperty mapper) {
        @SuppressWarnings("unchecked")
        TypeCodec<T> customCodec = (TypeCodec<T>) mapper.mappedProperty.getCustomCodec();
        if (value == null)
            bs.unset(i);
        else if (customCodec != null)
            bs.set(i, value, customCodec);
        else
            bs.set(i, value, mapper.mappedProperty.getPropertyType());
    }

    private static EnumMap<Option.Type, Option> toMap(Option[] options) {
        EnumMap<Option.Type, Option> result = new EnumMap<>(Option.Type.class);
        for (Option option : options) {
            result.put(option.type, option);
        }
        return result;
    }

    private void checkNotInEventLoop() {
        Session session = manager.getSession();
        if (session instanceof AbstractSession) {
            ((AbstractSession) session).checkNotInEventLoop();
        }
    }
}
